package com.example.kafka;

import java.util.Collections;
import java.util.Properties;

import com.example.config.KafkaProperties;
import com.example.exception.OpcException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;

/**
 * Copyright: Copyright (c) 2022
 *
 * <p>
 * Description:
 *
 * @author LiuFeng
 * @version 2022/11/14 10:42
 * @since
 */
@Service
@Slf4j
public class PCLKafkaService implements ApplicationRunner {

    @Autowired
    private KafkaProperties kafkaProperties;

    private static volatile KafkaProducer<String, String> producer;
    private static volatile KafkaConsumer<String, String> consumer;

    /**
     * 项目启动时，自动连接Kafka的 broker 和 consumer
     *
     * @param args incoming application arguments
     */
    @Override
    public void run(ApplicationArguments args) {
        producer = getProducer();
        consumer = getConsumer();
    }

    /**
     * 获取producer
     * @return
     */
    private KafkaProducer<String, String> getProducer() {
        KafkaProducer<String, String> producer = null;

        if (producer == null) {
            synchronized (PCLKafkaService.class) {
                if (producer == null) {
                    Properties props = new Properties();
                    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBroker());
                    //等待leader将数据写入本地log，即可发送下一条消息
                    props.put(ProducerConfig.ACKS_CONFIG, "1");
                    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
                    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

                    //消息压缩
//                    props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");

//                    props.put(ProducerConfig.BATCH_SIZE_CONFIG, "1");
//                    props.put(ProducerConfig.LINGER_MS_CONFIG, "1");

                    //SASL 认证参数设置
                    props.put("security.protocol", "SASL_TBDS");
                    props.put("sasl.mechanism", "TBDS");
                    props.put("sasl.tbds.secure.id", kafkaProperties.getAuthId());
                    props.put("sasl.tbds.secure.key", kafkaProperties.getAuthKey());


                    //创建topic
//                    NewTopic newTopic = new NewTopic(kafkaProperties.getTopic(), 3, (short) 1);
//                    AdminClient adminClient = AdminClient.create(props);
//                    CreateTopicsResult topics = adminClient.createTopics(Lists.newArrayList(newTopic));

                    producer = new KafkaProducer<>(props);
                    close();
                    log.info("producer连接到Kafka的Broker，broker是：{}", kafkaProperties.getBroker());
                }
            }
        }
        return producer;
    }

    /**
     * 获取consumer
     * @return
     */
    private KafkaConsumer<String, String> getConsumer() {
        KafkaConsumer<String, String> consumer = null;
        if (consumer == null) {
            synchronized (PCLKafkaService.class) {
                if (consumer == null) {

                    Properties props = new Properties();
                    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBroker());
                    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
                    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
                    props.put(ConsumerConfig.GROUP_ID_CONFIG, "test_group_001");

                    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaProperties.getOffset());
                    // props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");


                    props.put("security.protocol", "SASL_TBDS");
                    props.put("sasl.mechanism", "TBDS");
                    props.put("sasl.tbds.secure.id", kafkaProperties.getAuthId());
                    props.put("sasl.tbds.secure.key", kafkaProperties.getAuthKey());

                    // SASL/PLAIN认证参数设置
                    // props.put("security.protocol", "SASL_PLAINTEXT");
                    // props.put("sasl.mechanism", "PLAIN");


                    consumer = new KafkaConsumer<>(props);
                }
            }
        }
        return consumer;
    }

    /**
     * 项目停止时，关闭连接
     */
    private void close() {
        Runtime runtime = Runtime.getRuntime();
        runtime.addShutdownHook(new Thread() {
            @Override
            public void run() {
                super.run();
                if (producer != null) {
                    producer.close();
                }
            }
        });
    }

    /**
     * 发送消息到Kafka
     *
     * @param message 消息内容
     */
    public void sendMessage(String message) {
        String topic = kafkaProperties.getTopic();
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
        try {
            producer.send(record, (recordMetadata, exception) -> {
                if (exception != null) {
                    log.error("消息发送失败：{}", exception.getMessage());
                }
                if (recordMetadata != null) {
                    log.info("消息发送成功：topic={}, partition={}, offset={}", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
                }
            });
        } catch (Exception exception) {
            log.error("发送消息失败，原因：{}", exception.getMessage());
            throw new OpcException(exception);
        }
    }

    public void receiveMessage() {
        String topic = kafkaProperties.getTopic();
        consumer.subscribe(Collections.singletonList(topic));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(10000);
            for (ConsumerRecord<String, String> record : records) {
                log.info("Received message: offset={}, value={}", record.offset(), record.value());
            }
        }

    }

    public void deleteMessage() {
//        AdminClient adminClient = AdminClient.create(props);
    }
}
